package io.kiki.sba.registry.server.data_store.multi.cluster.executor;

import io.kiki.sba.registry.server.data_store.bootstrap.MultiClusterDataServerConfig;
import io.kiki.sba.registry.task.KeyedThreadPoolExecutor;
import io.kiki.sba.registry.task.MetricsableThreadPoolExecutor;
import io.kiki.sba.registry.util.NamedThreadFactory;
import lombok.Getter;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;


public class MultiClusterExecutorManager {

    private static final String REMOTE_SYNC_LEADER_EXECUTOR = "REMOTE_SYNC_LEADER_EXECUTOR";
    private static final String REMOTE_SYNC_DATA_ID_EXECUTOR = "REMOTE_SYNC_DATA_ID_EXECUTOR";
    private static final String REMOTE_DATA_CHANGE_REQUEST_EXECUTOR = "REMOTE_DATA_CHANGE_REQUEST_EXECUTOR";
    private static final String REMOTE_SLOT_SYNC_PROCESSOR_EXECUTOR = "REMOTE_SLOT_SYNC_PROCESSOR_EXECUTOR";
    @Getter
    private final KeyedThreadPoolExecutor remoteSyncLeaderExecutor;
    @Getter
    private final KeyedThreadPoolExecutor remoteSyncDataIdExecutor;
    @Getter
    private final ThreadPoolExecutor remoteDataChangeRequestExecutor;
    @Getter
    private final MetricsableThreadPoolExecutor remoteSlotSyncProcessorExecutor;
    private Map<String, KeyedThreadPoolExecutor> reportExecutors = new HashMap<>();

    private Map<String, MetricsableThreadPoolExecutor> metricsableExecutors = new HashMap<>();

    public MultiClusterExecutorManager(MultiClusterDataServerConfig multiClusterDataServerConfig) {
        remoteSyncLeaderExecutor = reportExecutors.computeIfAbsent(REMOTE_SYNC_LEADER_EXECUTOR, k -> new KeyedThreadPoolExecutor(REMOTE_SYNC_LEADER_EXECUTOR, multiClusterDataServerConfig.getRemoteSyncSlotLeaderExecutorThreadSize(), multiClusterDataServerConfig.getRemoteSyncSlotLeaderExecutorQueueSize()));

        remoteSyncDataIdExecutor = reportExecutors.computeIfAbsent(REMOTE_SYNC_DATA_ID_EXECUTOR, k -> new KeyedThreadPoolExecutor(REMOTE_SYNC_DATA_ID_EXECUTOR, multiClusterDataServerConfig.getRemoteSyncDataIdExecutorThreadSize(), multiClusterDataServerConfig.getRemoteSyncDataIdExecutorQueueSize()));

        remoteDataChangeRequestExecutor = metricsableExecutors.computeIfAbsent(REMOTE_DATA_CHANGE_REQUEST_EXECUTOR, k -> new MetricsableThreadPoolExecutor(REMOTE_DATA_CHANGE_REQUEST_EXECUTOR, multiClusterDataServerConfig.getRemoteDataChangeExecutorThreadSize(), multiClusterDataServerConfig.getRemoteDataChangeExecutorThreadSize(), 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(multiClusterDataServerConfig.getRemoteDataChangeExecutorQueueSize()), new NamedThreadFactory(REMOTE_DATA_CHANGE_REQUEST_EXECUTOR, true)));

        remoteSlotSyncProcessorExecutor = metricsableExecutors.computeIfAbsent(REMOTE_SLOT_SYNC_PROCESSOR_EXECUTOR, k -> new MetricsableThreadPoolExecutor("RemoteSlotSyncProcessorExecutor", multiClusterDataServerConfig.getRemoteSlotSyncRequestExecutorMinPoolSize(), multiClusterDataServerConfig.getRemoteSlotSyncRequestExecutorMaxPoolSize(), 300, TimeUnit.SECONDS, new ArrayBlockingQueue<>(multiClusterDataServerConfig.getRemoteSlotSyncRequestExecutorQueueSize()), new NamedThreadFactory("RemoteSlotSyncProcessorExecutor", true)));
    }


}
